package vip.huhailong.redismq.redistool;


import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.UUID;

/**
 */
@Component
public class RedisPendingMessageHandler {

    private static final Logger log = LoggerFactory.getLogger(RedisPendingMessageHandler.class);
    @Autowired
    private RedisTemplate redisTemplate;

    @Value("${redis-stream.names}")
    private String[]redisStreamNames;               //redis stream 数组
    @Value("${redis-stream.groups}")
    private String[]groups;                         //redis stream 群组数组


    //@Scheduled(cron = "0/30 * * * * ?")
    public void processMessage() {
        for (String streamName : redisStreamNames) {
            StreamOperations sop = redisTemplate.opsForStream();
            List<MapRecord<String, Object, Object>> list = sop.read(Consumer.from(groups[0], this.getClass().getName()), StreamOffset.create(streamName, ReadOffset.lastConsumed()));
            for (MapRecord<String, Object, Object> message : list) {
                log.info("接收消息:[{}]", JSON.toJSONString(message));
                //sop.acknowledge(order_stream, order_group, message.getId());
            }
        }

    }


}
